package com.liy.framework.common.kafka;

import com.liy.framework.common.kafka.vo.OperationLog;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class MessageBox {

    @Value("${kafka.topics.logs:logsTopic}")
    private String logsTopicName;

    private final KafkaTemplate<String, String> kafkaTemplate;

    private Gson gson = (new GsonBuilder()).create();

    public MessageBox(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendLogsMsg(OperationLog message) {
        this.kafkaTemplate.send(this.logsTopicName, message.getUserId(), this.gson.toJson(message));
    }

    public void sendMsg(String topicName, String key, Object data) {
        this.kafkaTemplate.send(topicName, key, this.gson.toJson(data));
    }

    public void sendMsg(String topicName, Object data) {
        this.kafkaTemplate.send(topicName, this.gson.toJson(data));
    }

    public void sendMsg(String topicName, String data) {
        this.kafkaTemplate.send(topicName, data);
    }
}
